// -*-mode:c++; coding:utf-8-*-

#include "global.hpp"
#include "service_provider.hpp"

#include "service.hpp"

namespace xbase
{
	void task::run()
	{
		bool need_heartbeat = false;
		switch(m_request->get_header().m_type)
		{
			// master request
		case LOAD_TABLET_REQ:
			m_response.reset(response::factory_type::construct());
			global::the_service->do_load_tablet(m_request, m_response);
			need_heartbeat = true;
			break;
		case REMOVE_TABLET_REQ:
			m_response.reset(response::factory_type::construct());
			global::the_service->do_remove_tablet(m_request, m_response);
			break;
		case ROOT_META_RESP:
			global::the_service->do_root_meta(m_request);
			break;
		case TABLE_INFO_RESP:
			global::the_service->do_table_info(m_request);
			break;
		case TABLET_SERVER_RESTART:
			global::the_service->do_restart(m_request);
			break;

			// client request
		case READ_REQ:
			m_response.reset(response::factory_type::construct());
			global::the_service->do_read(m_request, m_response);
			break;
		case WRITE_REQ:
			m_response.reset(response::factory_type::construct());
			global::the_service->do_write(m_request, m_response);
			break;
		case BATCH_WRITE_REQ:
			m_response.reset(response::factory_type::construct());
			global::the_service->do_batch_write(m_request, m_response);
			break;
		case SCAN_OPEN_REQ:
			m_response.reset(response::factory_type::construct());
			global::the_service->do_scan_open(m_request, m_response);
			break;
		case SCAN_NEXT_REQ:
			m_response.reset(response::factory_type::construct());
			global::the_service->do_scan_next(m_request, m_response);
			break;
		case SCAN_CLOSE_REQ:
			m_response.reset(response::factory_type::construct());
			global::the_service->do_scan_close(m_request, m_response);
			break;
		case LIST_TABLET_REQ:
			m_response.reset(response::factory_type::construct());
			global::the_service->do_list_tablet(m_request, m_response);
			break;

			// unexpected
		default:
			global::the_service->do_unexpected(m_request);
			m_session->stop(); // stop this session
			break;
		}

		if(m_response)	// if has
		{
#ifndef USE_ASYNC_WRITE
			// 使用同步写，性能应该更好一些，但需要对socket加锁
			boost::system::error_code ec_ignore;
			m_session->write(m_response, ec_ignore);
			if(ec_ignore)
			{
				LOG(INFO, "task::run " << ec_ignore.message());
			}
#else
			// 使用异步写
			m_session->async_write_response(m_response);
#endif	// USE_ASYNC_WRITE

			// for stats
			// 注意：该值不准确，主要参照接受的请求数即可
			++ global::the_load_info->m_output_message_number;

			if(need_heartbeat)
			{
				response::pointer msg(response::factory_type::construct());
				global::the_service->construct_heartbeat(msg);
				LOG_ENV(const int ret = )
					global::the_master_session->heartbeat(msg);
				LOG_IF_ELSE((ret < 0),
					    WARN, "additional heartbeat error",
					    TRACE, "additional heartbeat done");
			}
		}
	}

	// response::pointer master_session::get_heartbeat_message()
	// {
	// 	// 必须创建一个新的，旧的会在适当的时候释放。因为旧的可能还会在其他线程中没使用完
	// 	response::pointer msg(response::factory_type::construct());

	// 	// if (! fs::exists(global::the_tablet_server_path)) {
	// 	// 	LOG(WARN, "server dir didn't exists,"
	// 	// 	    << " so quit and reset the server.");
	// 	// }

	// 	global::the_service->construct_heartbeat(msg);
	// 	return msg;
	// }

	static inline
	void handle_report_helper(response::pointer report, // 必须有该指针
				  bool *report_flag,
				  const boost::system::error_code &error)  {
		(void)report;	// 执行完后才释放

		if(! error)
		{
			*report_flag = true;
			LOG(INFO, "client::handle_report report sent");
		}
		else
		{
			LOG(ERROR, "client::handle_report " << error.message());
		}
	}

	void send_report_helper(session::pointer new_session,
				bool *report_flag) {
		response::pointer report(response::factory_type::construct());
		global::the_service->construct_report(report);
		new_session->async_write_response(report,
						  boost::bind(&handle_report_helper,
							      report, // 必须持有该指针
							      report_flag,
							      boost::asio::placeholders::error));

		// 再查询根子表位置
		global::the_service->query_root_meta_address();
	}

	void register_session_helper(session::pointer new_session) {
		global::the_master_session->register_session(new_session);
	}

	// void master_session::handle_heartbeat(session::pointer s,
	// 				      response::pointer heartbeat,
	// 				      const boost::system::error_code &error) {
	// 	(void)heartbeat;
	// 	if(!error)
	// 	{
	// 		LOG(INFO, "master_session::handle_heartbeat success");
	// 		boost::mutex::scoped_lock lock(m_mutex);
	// 		m_heartbeat_timer->expires_at(m_heartbeat_timer->expires_at() + boost::posix_time::seconds(global::the_conf->heart_beat_interval));
	// 		m_heartbeat_timer->async_wait(boost::bind(&master_session::heartbeat,
	// 							 this,
	// 							 s,
	// 							 boost::asio::placeholders::error));
	// 	}
	// 	else
	// 	{
	// 		LOG(ERROR, "master_session::handle_heartbeat " << error.message());
	// 		// 发送心跳失败，则将连接关闭，将指针清空，使其建立新的连接。
	// 		clear_session();
	// 	}
	// }
	
	void session::handle_request(request::pointer req,
				     const boost::system::error_code& error,
				     std::size_t bytes_transferred) {
		if(error)
		{
			LOG_IF_ELSE((error == boost::asio::error::eof), // eof记录INFO日志
				    INFO, "session::handle_request connection closed",
				    ERROR, "session::handle_request " << error.message());
			return;
		}
		// assert(bytes_transferred == req->data_length());
	
		task t(pointer(this), req);
		m_task_runner.add_task(t);
	
		// read next request
		async_read_request();
	}

	void task_runner::add_task(task t) {
		{
			boost::mutex::scoped_lock lock(m_mutex);
			m_queue.push(t);
		
			++ global::the_load_info->m_input_message_number;
			++ global::the_load_info->m_left_input_message_number;
		
			++ m_counter;
		}
		m_cond.notify_one();
	}

	void task_runner::runner() {
		while(true)
		{
			task the_task;
			{
				boost::mutex::scoped_lock lock(m_mutex);
				if(m_stop) // stopped
				{
					LOG(INFO, "task_runner::runner stopped");
					return;
				}
				
				while(m_queue.empty())
				{
					m_cond.wait(lock);
					
					if(m_stop) // stopped while waiting
					{
						LOG(INFO, "task_runner::runner stopped");
						return;
					}
				}
				the_task = m_queue.front();
				m_queue.pop();
				
				// for status
				-- global::the_load_info->m_left_input_message_number;
			}
			the_task.run();
		}
	}

}
